package org.jeecg.modules.friends.base;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;

import java.util.function.Supplier;

import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.MDC;

import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import lombok.extern.slf4j.Slf4j;

/**
 * Created on 2022-10-22
 */
@Slf4j
public class ThreadPoolTool {
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            10, // 初始化线程数大小
            30,// 最大核心线程大小
            1, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(2000), // 阻塞队列
            new ThreadFactoryBuilder().setNameFormat("ThreadPoolUtil-%d").build()); // 给每个线程命名

    private static ThreadPoolExecutor threadPoolExecuteParallel = new ThreadPoolExecutor(
            10, // 初始化线程数大小
            50,// 最大核心线程大小
            1, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), // 阻塞队列
            new ThreadFactoryBuilder().setNameFormat("PigExecuteParallel ThreadPool-%d").build()); // 给每个线程命名

    public static void execute(Runnable runnable) {
        threadPoolExecutor.execute(() -> {
            try {
                MDC.put("TRACE_ID", UUID.randomUUID().toString());
                runnable.run();
            } catch (Exception e) {
                log.error("error,", e);
                NotifyTool.weChatAlarm(e);
            }
        });
    }

    public static void executeFailure(Runnable runnable, Consumer<Exception> consumer) {
        threadPoolExecutor.execute(() -> {
            try {
                MDC.put("TRACE_ID", UUID.randomUUID().toString());
                runnable.run();
            } catch (Exception e) {
                log.error("error,", e);
                NotifyTool.weChatAlarm(e);
                consumer.accept(e);
            }
        });
    }

    /**
     * 多线程批量处理数据
     *
     * @param maxId 表数据的最大ID
     * @param perBatchCount 每个现成处理的数据量
     */
    public static void batchProcess(long maxId, long perBatchCount, Consumer<Pair<Long, Long>> consumer) {
        List<Pair<Long, Long>> pairs = generateSegment(maxId, perBatchCount);
        for (Pair<Long, Long> pair : pairs) {
            ThreadPoolTool.execute(() -> {
                log.info("批次[{},{}]", pair.getLeft(), pair.getRight());
                consumer.accept(pair);
            });

        }
    }

    /**
     * 生成批次范围
     */
    private static List<Pair<Long, Long>> generateSegment(long total, long perBatchCount) {
        long totalBatch = total % perBatchCount == 0 ? total / perBatchCount : total / perBatchCount + 1;
        long leftId;
        long rightId = 0;
        List<Pair<Long, Long>> pairs = com.google.common.collect.Lists.newArrayList();
        for (long i = 1; i <= totalBatch; i++) {
            leftId = rightId + 1;
            rightId = i * perBatchCount;
            pairs.add(Pair.of(leftId, rightId));
        }
        return pairs;
    }

    /**
     * 并行执行
     *
     * @param <REQ> 请求对象
     * @param <A> 返回结果 MAP<String,A> String:为 requestList 对象的toString 方法
     */
    public static <REQ, A> Map<String, A> executeParallel(List<REQ> requestList, Function<REQ, A> function) {
        Map<String, Future<A>> futureMap = Maps.newHashMap();
        Map<String, A> resultFutureMap = Maps.newHashMap();
        for (REQ r : requestList) {
            Future<A> future = threadPoolExecuteParallel.submit(() -> {
                A apply = null;
                try {
                    MDC.put("TRACE_ID", UUID.randomUUID().toString());
                    apply = function.apply(r);
                } catch (Exception e) {
                    log.error("executeParallel error,", e);
                    NotifyTool.weChatAlarm(e);
                }
                return apply;
            });
            futureMap.put(r.toString(), future);
        }
        futureMap.forEach((key, value) -> {
            try {
                resultFutureMap.put(key, value.get());
            } catch (Exception e) {
                log.error("error forEach get ,", e);
                NotifyTool.weChatAlarm(e);
            }
        });
        return resultFutureMap;
    }

    /**
     * @param <R> 返回结果
     */
    public static <R> Future<R> executeAsync(Supplier<R> supplier) {
        return threadPoolExecuteParallel.submit(() -> {
            MDC.put("TRACE_ID", UUID.randomUUID().toString());
            return supplier.get();
        });
    }
}
